package org.project.pvuv

import java.util.Properties

import org.FlinkStreamApp
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.project.bean.UserBebavior
import redis.clients.jedis.Jedis

/**
 * description ：使用布隆过滤器 计算 uv
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/27 8:20
 * modified By ：
 * version:    : 1.0
 */
object UvWithBloomFilter extends FlinkStreamApp {

  val uvTable = "uvTable"

  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // kafka 连接配置
    val props = new Properties()
    props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092")
    props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my-group")
    props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
    props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")

    // 读取 kafka 数据
    val source = environment.addSource(new FlinkKafkaConsumer[String]("t01", new SimpleStringSchema(), props))

    // 计算 uv
    source
      .map(line => {
        val splits = line.split(",")
        UserBebavior(splits(0).toLong, splits(1).toLong, splits(2).toInt, splits(3), splits(4).toLong * 1000)
      })
      .filter(_.behavior == "pv")
      .assignAscendingTimestamps(_.timestamp)
      .map(ub => ("uv", ub.userId))
      .keyBy(_._1)
      .timeWindow(Time.minutes(60), Time.minutes(5))
      .trigger(new MyTrigger)
      .process(new MyProcess)
      .print
  }

  // 布隆过滤器
  class BloomFilter(size: Int) extends Serializable {
    def getHashOffset(key: String): Long = {
      var result = 0
      for (i <- 0 until key.length) {
        result = result * 61 + key.charAt(i)
      }
      (size - 1) & result
    }
  }

  class MyProcess extends ProcessWindowFunction[(String, Long), String, String, TimeWindow] {
    // 初始化布隆过滤器
    lazy val bloomFilter = new BloomFilter(1 << 20)

    // 初始化 redis 连接
    lazy val redisClient = new Jedis("hadoop101", 6379)

    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[String]) = {
      // 来一个元素，就通过布隆过滤器计算它的 offset, 由于触发器，每次执行窗口并清空数据，所以全窗口函数中，最后一个元素就是当前触发器触发的元素
      val hashOffset = bloomFilter.getHashOffset("" + elements.last._2)

      // 写入 redis 中的 bit 数组，如果存在就不写入了，如果不存在就将 uv 数量 +1
      val bitKey = "uvBloom"
      if (!redisClient.getbit(bitKey, hashOffset)) {
        // 设置 bit 数组值为 true
        redisClient.setbit(bitKey, hashOffset, true)
        // uv 数量加1 key 当前窗口关闭时间
        val endKey = "" + context.window.getEnd
        val uvTableValue = redisClient.hget(uvTable, endKey)
        var uvCount = 0L
        if (null != uvTableValue) {
          uvCount = uvTableValue.toLong
        }
        redisClient.hset(uvTable, endKey, (uvCount + 1).toString)
      }
    }
  }

  class MyTrigger extends Trigger[(String, Long), TimeWindow] {
    lazy val redisClinet = new Jedis("hadoop101", 6379)

    // 每个元素进来一次
    override def onElement(element: (String, Long), timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext) = {
      ctx.registerEventTimeTimer(window.getEnd)
      TriggerResult.FIRE_AND_PURGE
    }

    // 定时器被触发，定时器被触发说明，已经到了窗口关闭时间，就获取 redis 中的 uv 数量
    override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext) = {
      //      println(s"水位线${ctx.getCurrentWatermark},  time = $time, window = ${window.getEnd}")
      if (time >= window.getEnd) {
        val uvCount = redisClinet.hget(uvTable, window.getEnd.toString)
        if (null != uvCount) {
          println(s"窗口时间 ${window.getEnd} uv = $uvCount")
        }
      }
      TriggerResult.CONTINUE
    }

    // 被系统定时器触发了，什么都不做
    override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext) = {
      TriggerResult.CONTINUE
    }

    // 清除工作
    override def clear(window: TimeWindow, ctx: Trigger.TriggerContext) = {
    }
  }

}
